package com.ada.flink

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._

object DataSetWcApp {

    def main(args: Array[String]): Unit = {

        val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
        val inputPath: String = parameterTool.get("input")
        val outputPath: String = parameterTool.get("output")

        // 1、env
        val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

        // 2、source
        val txtDS: DataSet[String] = env.readTextFile(inputPath)

        // 3、transform
        val aggDS: AggregateDataSet[(String, Int)] = txtDS.flatMap(_.split(" ")).map((_, 1)).groupBy(0).sum(1)

        // 4、sink
        aggDS.writeAsCsv(outputPath).setParallelism(1)

        //aggDS.print()

        env.execute()
    }

}
